package com.atguigu.edu.realtime.app.dim.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.constant.EduConfig;
import com.atguigu.edu.realtime.constant.EduConstant;
import com.atguigu.edu.realtime.util.PhoenixUtil;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Map;
import java.util.Set;

public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    // 将数据组合成SQL，然后调用PhoenixUtil写入phoenix
    public void invoke(JSONObject jsonObject, Context context) throws Exception {
        //1.把jsonObject解析为一个sql upsert
        //upsert into ${sink_table}(字段名1.,,,,,,,) value(字段值1.,,,, )
        //提取data的键值对
        StringBuilder fieldNameStringBuilder = new StringBuilder("(");//字段
        StringBuilder fieldValueStringBuilder = new StringBuilder("(");//字段值

        //提取data
        JSONObject datajsonObj = jsonObject.getJSONObject("data");
        //提取键值对
        Set<Map.Entry<String, Object>> entries = datajsonObj.entrySet();
        int count = 0;
        for (Map.Entry<String, Object> entry : entries) {//得到键值对集合
            String fieldName = entry.getKey();
            Object fieldValue = entry.getValue();

            if (fieldNameStringBuilder.length() > 1){
                fieldNameStringBuilder.append(",");
                fieldValueStringBuilder.append(",");
            }
            fieldNameStringBuilder.append(fieldName);
            fieldValueStringBuilder.append("'"+fieldValue+"'");
        }
        fieldNameStringBuilder.append(")");
        fieldValueStringBuilder.append(")");

        //表名提取出来
        String sinkTable = jsonObject.getString("sink_table");
        StringBuilder upsertSQL = new StringBuilder("upsert into " + EduConfig.PHOENIX_SCHEMA+"."+sinkTable)
                .append(fieldNameStringBuilder).append("values").append(fieldValueStringBuilder);

    //    System.out.println(upsertSQL);
        //2.把upsert交给PhoenixUtil执行
        PhoenixUtil.executeSql(upsertSQL.toString());
    }
}
